package com.zk.storm;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class DataAggregationTopology {

    public static void main(String[] args) throws Exception {
        // 1. 配置KafkaSpout
        KafkaSpoutConfig<String, String> kafkaConfig = KafkaSpoutConfig.builder("localhost:9092", "test")
                .setProp(ConsumerConfig.GROUP_ID_CONFIG, "log-analysis-group")
                .setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest") //只消费最新产生的数据
                .build();

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("kafka-spout", new KafkaSpout<>(kafkaConfig), 1); // 单并行度测试

        // 2. IP地理位置解析Bolt（从KafkaSpout接收原始消息）
        builder.setBolt("ip-location-bolt", new IpLocationBolt(), 2)
                .shuffleGrouping("kafka-spout"); // 随机分发Kafka消息到地理位置Bolt

        // 3. 数据解析Bolt（从ip-location-bolt获取原始消息）
        builder.setBolt("parse-bolt", new DataParseBolt(), 2)
                .shuffleGrouping("ip-location-bolt"); // 随机分发地理位置解析后的消息

        // 业务需求1：每日国家总点击量
        builder.setBolt("country-sum-bolt", new DailyCountrySumBolt(), 2)
                .fieldsGrouping("parse-bolt", new Fields("date_key", "country"));

        // 业务需求2：每日系统平均点击量
        builder.setBolt("platform-avg-bolt", new DailyPlatformAvgBolt(), 2)
                .fieldsGrouping("parse-bolt", new Fields("date_key", "platform"));

        // 业务需求3：每日设备最大点击量
        builder.setBolt("device-max-bolt", new DailyDeviceMaxBolt(), 2)
                .fieldsGrouping("parse-bolt", new Fields("date_key", "device"));

        // 业务需求4：每日浏览器总点击量
        builder.setBolt("agent-sum-bolt", new DailyAgentSumBolt(), 2)
                .fieldsGrouping("parse-bolt", new Fields("date_key", "agent"));


        // 5. HBase写入Bolt
        builder.setBolt("hbase-write-bolt", new HBaseWriteBolt(), 2)
                .allGrouping("country-sum-bolt")
                .allGrouping("platform-avg-bolt")
                .allGrouping("device-max-bolt")
                .allGrouping("agent-sum-bolt");

        // 启动本地拓扑
        Config config = new Config();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("multi-agg-topo", config, builder.createTopology());

        // 定时停止
        int stopAfterSeconds = 180; // 3分钟
        Thread.sleep(stopAfterSeconds * 1000);

        // 停止拓扑并关闭集群
        cluster.killTopology("multi-agg-topo");
        cluster.shutdown();
        System.out.println("拓扑已自动停止");
    }
}